package test.top1.com.atguigu.app.dwd.db;

import com.sun.xml.internal.bind.v2.TODO;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import test.top1.com.atguigu.utils.MyKafkaUtil;
import test.top1.com.atguigu.utils.MySqlUtil;

import java.time.Duration;

/**
 * ClassName: Dwd03_TradeOrderDetail
 * Package: test.top1.com.atguigu.app.dwd.db
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/4 10:26
 * @Version 1.0
 */
public class Dwd03_TradeOrderDetail {
    public static void main(String[] args) {

        //TODO 1.获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //设置ttl
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        //TODO 生产环境参数
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 2.从kafka读取 topic_db数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_order_detail_221109"));

        //TODO 3.过滤出订单明细数据
        Table orderDetailTable = tableEnv.sqlQuery("" +
                "select \n" +
                "    `data`[`id`] id, \n" +
                "    `data`[`order_id`] order_id, \n" +
                "    `data`[`sku_id`] sku_id, \n" +
                "    `data`[`sku_name`] sku_name, \n" +
                "    `data`[`img_url`] img_url, \n" +
                "    `data`[`order_price`] order_price, \n" +
                "    `data`[`sku_num`] sku_num, \n" +
                "    `data`[`create_time`] create_time, \n" +
                "    `data`[`source_type`] source_type, \n" +
                "    `data`[`source_id`] source_id, \n" +
                "    `data`[`split_total_amount`] split_total_amount, \n" +
                "    `data`[`split_activity_amount`] split_activity_amount, \n" +
                "    `data`[`split_coupon_amount`] split_coupon_amount, \n" +
                "    `data`[`operate_time`] operate_time, \n" +
                "    `pt`\n" +
                "from topic_db\n" +
                "where `database`=`gmall`\n" +
                "and `table`=`order_detail`\n" +
                "and `type`=`insert`");
        tableEnv.createTemporaryView("order_detail",orderDetailTable);

        //TODO 4.过滤出订单数据
        Table orderInfoTable = tableEnv.sqlQuery("" +
                "select \n" +
                "    `data`[`id`] id,\n" +
                "    `data`[`consignee`] consignee,\n" +
                "    `data`[`consignee_tel`] consignee_tel,\n" +
                "    `data`[`total_amount`] total_amount,\n" +
                "    `data`[`order_status`] order_status,\n" +
                "    `data`[`user_id`] user_id,\n" +
                "    `data`[`payment_way`] payment_way,\n" +
                "    `data`[`delivery_address`] delivery_address,\n" +
                "    `data`[`order_comment`] order_comment,\n" +
                "    `data`[`out_trade_no`] out_trade_no,\n" +
                "    `data`[`trade_body`] trade_body,\n" +
                "    `data`[`create_time`] create_time,\n" +
                "    `data`[`operate_time`] operate_time,\n" +
                "    `data`[`expire_time`] expire_time,\n" +
                "    `data`[`process_status`] process_status,\n" +
                "    `data`[`tracking_no`] tracking_no,\n" +
                "    `data`[`parent_order_id`] parent_order_id,\n" +
                "    `data`[`img_url`] img_url,\n" +
                "    `data`[`province_id`] province_id,\n" +
                "    `data`[`activity_reduce_amount`] activity_reduce_amount,\n" +
                "    `data`[`coupon_reduce_amount`] coupon_reduce_amount,\n" +
                "    `data`[`original_total_amount`] original_total_amount,\n" +
                "    `data`[`feight_fee`] feight_fee,\n" +
                "    `data`[`feight_fee_reduce`] feight_fee_reduce,\n" +
                "    `data`[`refundable_time`] refundable_time\n" +
                "from topic_db\n" +
                "where `database`=`gmall`\n" +
                "and `table`=`order_info`\n" +
                "and `type`=`insert`");
        tableEnv.createTemporaryView("order_info",orderInfoTable);

        //TODO 5.过滤出订单明细活动数据
        Table orderActivityTable = tableEnv.sqlQuery("" +
                "select \n" +
                "    `data`[`id`] id,\n" +
                "    `data`[`order_id`] order_id,\n" +
                "    `data`[`order_detail_id`] order_detail_id,\n" +
                "    `data`[`activity_id`] activity_id,\n" +
                "    `data`[`activity_rule_id`] activity_rule_id,\n" +
                "    `data`[`sku_id`] sku_id,\n" +
                "    `data`[`create_time`] create_time,\n" +
                "    `data`[`operate_time`] operate_time\n" +
                "from topic_db\n" +
                "where `database`=`gmall`\n" +
                "and `table`=`order_detail_activity`\n" +
                "and `type`=`insert`");
        tableEnv.createTemporaryView("order_activity", orderActivityTable);

        //TODO 6.过滤出订单明细优惠券数据
        Table orderCouponTable = tableEnv.sqlQuery("" +
                "select \n" +
                "   `data`[`id`] id,\n" +
                "   `data`[`order_id`] order_id,\n" +
                "   `data`[`order_detail_id`] order_detail_id,\n" +
                "   `data`[`coupon_id`] coupon_id,\n" +
                "   `data`[`coupon_use_id`] coupon_use_id,\n" +
                "   `data`[`sku_id`] sku_id,\n" +
                "   `data`[`create_time`] create_time,\n" +
                "   `data`[`operate_time`] operate_time\n" +
                "from topic_db\n" +
                "where `database`=`gmall`\n" +
                "and `table`=`order_detail_coupon`\n" +
                "and `type`=`insert`");
        tableEnv.createTemporaryView("order_coupon", orderCouponTable);

        //TODO 7.读取MySQL base_dic表
        tableEnv.executeSql(MySqlUtil.getMysqlBaseDicDDL());

        //TODO 8.五表关联
        Table resultTable = tableEnv.sqlQuery("" +
                "select " +
                "    od.id, " +
                "    od.order_id, " +
                "    od.sku_id, " +
                "    od.sku_name, " +
                "    od.order_price, " +
                "    od.sku_num, " +
                "    od.create_time, " +
                "    od.split_total_amount, " +
                "    od.split_activity_amount, " +
                "    od.split_coupon_amount, " +
                "    oi.consignee, " +
                "    oi.consignee_tel, " +
                "    oi.total_amount, " +
                "    oi.order_status, " +
                "    oi.user_id, " +
                "    oi.payment_way, " +
                "    oi.delivery_address, " +
                "    oi.order_comment, " +
                "    oi.out_trade_no, " +
                "    oi.trade_body, " +
                "    oi.process_status, " +
                "    oi.tracking_no, " +
                "    oi.parent_order_id, " +
                "    oi.province_id, " +
                "    oi.activity_reduce_amount, " +
                "    oi.coupon_reduce_amount, " +
                "    oi.original_total_amount, " +
                "    oi.feight_fee, " +
                "    oi.feight_fee_reduce, " +
                "    oi.refundable_time, " +
                "    oa.activity_id, " +
                "    oa.activity_rule_id, " +
                "    oc.coupon_id, " +
                "    oc.coupon_use_id, " +
                "    dic.dic_name " +
                "from order_detail od " +
                "join order_info oi " +
                "on od.order_id = oi.id " +
                "left join order_activity oa " +
                "on od.id = oa.order_detail_id " +
                "left join order_coupon oc " +
                "on od.id = oc.order_detail_id " +
                "join base_dic FOR SYSTEM_TIME AS OF od.pt AS dic " +
                "on oi.order_status=dic.dic_code");
        tableEnv.createTemporaryView("result_table", resultTable);

        //TODO 9.使用upsert-kafka创建DWD层下单主题
        tableEnv.executeSql("" +
                "CREATE TABLE dwd_order_detail (\n" +
                "    `id` string,\n" +
                "    `order_id` string,\n" +
                "    `sku_id` string,\n" +
                "    `sku_name` string,\n" +
                "    `order_price` string,\n" +
                "    `sku_num` string,\n" +
                "    `create_time` string,\n" +
                "    `split_total_amount` string,\n" +
                "    `split_activity_amount` string,\n" +
                "    `split_coupon_amount` string,\n" +
                "    `consignee` string,\n" +
                "    `consignee_tel` string,\n" +
                "    `total_amount` string,\n" +
                "    `order_status` string,\n" +
                "    `user_id` string,\n" +
                "    `payment_way` string,\n" +
                "    `delivery_address` string,\n" +
                "    `order_comment` string,\n" +
                "    `out_trade_no` string,\n" +
                "    `trade_body` string,\n" +
                "    `process_status` string,\n" +
                "    `tracking_no` string,\n" +
                "    `parent_order_id` string,\n" +
                "    `province_id` string,\n" +
                "    `activity_reduce_amount` string,\n" +
                "    `coupon_reduce_amount` string,\n" +
                "    `original_total_amount` string,\n" +
                "    `feight_fee` string,\n" +
                "    `feight_fee_reduce` string,\n" +
                "    `refundable_time` string,\n" +
                "    `activity_id` string,\n" +
                "    `activity_rule_id` string,\n" +
                "    `coupon_id` string,\n" +
                "    `coupon_use_id` string,\n" +
                "    `dic_name` string\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + MyKafkaUtil.getKafkaUpsertSinkConnOption("dwd_trade_order_detail"));

        //TODO 10.插入数据
        tableEnv.executeSql("insert into dwd_order_detail select * from result_table");

    }
}
